package com.company.messagequeue.kafka.producer;

import com.company.messagequeue.kafka.constant.KafkaConstant;
import com.company.messagequeue.message.CommonMessage;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.log4j.Logger;

import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Title: kafkaProducer<br>
 * Description: kafkaProducer<br>
 * CreateDate:2020年12月04日 19:13
 *
 */
public class KafkaProducerClient {
  private static Logger logger = Logger.getLogger(KafkaProducerClient.class);

  /**
   * 可能以后设置多个producer用一个map来存储,目前就用一个来发送
   */
  private static ConcurrentHashMap<String, Producer> producerMap = new ConcurrentHashMap();

  /**
   * Title: 获取名称获取生产者<br>
   * Description: 这边可以根据不同名称获取复数生产者，但是同一个名称只有一个生产者<br>
   * CreateDate: 2020年12月05日 16:25:22<br>
   *
   * @param producerName
   * @return org.apache.kafka.clients.producer.Producer
   * @throws Exception
   * @category getProducer
   */
  synchronized static private Producer getProducer(String producerName, String serverAddress) {
    Producer producer = producerMap.get(producerName);
    if (producer == null) {

      Properties props = new Properties();
      props.put("bootstrap.servers", serverAddress);
      props.put("acks", "all");
      // delivery.timeout.ms should be equal to or larger than linger.ms +
      // request.timeout.ms 需要看下每个参数的意思
      // props.put("delivery.timeout.ms", 300000);
      // props.put("batch.size", 16384);
      // props.put("linger.ms", 1);
      // props.put("buffer.memory", 33554432);
      props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      producer = new KafkaProducer<String, String>(props);

      producerMap.put(producerName, producer);
    }
    return producer;
  }

  /**
   * Title: 发送消息<br>
   * Description: <br>
   * CreateDate: 2020年12月05日 14:12:15<br>
   *
   * @param producerName
   * @param queueName
   * @param data
   * @param serverAddress
   * @return
   * @throws Exception
   * @category 发送消息
   */
  public static void send(String producerName, String queueName, Object data,
      String serverAddress) {
    logger.info("kafka消息发送开始======>queueName:" + queueName + "data:" + data);
    CommonMessage message = new CommonMessage(queueName, data);
    Producer producer = getProducer(producerName, serverAddress);
    producer.send(new ProducerRecord<String, String>(message.getQueueName(), null,
        message
            .toString()));

    // 为了能及时发出去,需要flush下
    producer.flush();
    logger.info("kafka消息发送结束");
  }

  /**
   * Title: 发送消息<br>
   * Description: <br>
   * CreateDate: 2020年12月05日 14:12:15<br>
   *
   * @param producerName
   * @param queueName
   * @param data
   * @return
   * @throws Exception
   * @category 发送消息
   */
  public static void send(String producerName, String queueName, Object data) {
    send(producerName, queueName, data, KafkaConstant.SERVER_ADDRESS);
  }

  /**
   * Title: 发送消息<br>
   * Description: <br>
   * CreateDate: 2020年12月05日 14:12:15<br>
   *
   * @param queueName
   * @param data
   * @return
   * @throws Exception
   * @category 发送消息
   */
  public static void send(String queueName, Object data,
      String serverAddress) {
    send(KafkaConstant.DEFAULT_PRODUCER_NAME, queueName, data, serverAddress);
  }

  /**
   * Title: 发送消息<br>
   * Description: <br>
   * CreateDate: 2020年12月05日 14:12:15<br>
   *
   * @param queueName
   * @param data
   * @return
   * @throws Exception
   * @category 发送消息
   */
  public static void send(String queueName, Object data) {
    send(KafkaConstant.DEFAULT_PRODUCER_NAME, queueName, data,
        KafkaConstant.SERVER_ADDRESS);
  }
}
